package my;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import function.CustomerDeserialization;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class flinkcdc {
    public static void main(String[] args) throws Exception {
        System.setProperty("HADOOP_USER_NAME","root");
        System.setProperty("HADOOP_USER_PASSWORD","zyb705");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //1.1 开启checkpoint，并指定状态后端为FS         memory fs rocksdb

        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://192.168.45.132:9000/flink/ck");
        env.enableCheckpointing(1000);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().enableExternalizedCheckpoints(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        //2.通过FlinkCDC构建SourceFunction并读取数据
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("192.168.45.132")
                .port(3306)
                .username("root")
                .password("123456")
                .databaseList("gmall_flink")
                .tableList("gmall_flink.base_trademark")   //如果不添加该参数，则消费指定数据库中所有表的数据，如果指定，指定方式为db.table
                .deserializer(new CustomerDeserialization())
//                .debeziumProperties()
                .startupOptions(StartupOptions.initial())   //initial(): 无checkPoint时，都是从头开始读;    latest(): 直接读最新的数据
                .build();


        DataStreamSource<String> streamSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");

        //3.打印数据，并将数据写入Kafka
        streamSource.print();
//        String sinkTopic = "ods_base_db";
//        streamSource.addSink(MyKafkaUtil.getKafkaProducer(sinkTopic));

        //4.启动任务
        env.execute("FlinkCDC");
    }
}
